package com.mongo.data.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.mongo.data.entity.NodeInfo;
import com.mongo.data.entity.MissionStatus;
import com.mongo.data.entity.UploadNode;
import com.mongo.data.enums.JobStatus;
import com.mongo.data.enums.NodeStatus;
import com.mongo.data.enums.NodeType;
import com.mongo.data.service.MissionStatusService;
import com.mongo.data.utils.JsonUtils;
import com.mongo.data.vo.*;
import com.mongodb.client.result.DeleteResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.*;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class MissionStatusServiceImpl implements MissionStatusService {

    @Autowired
    private MongoTemplate mongoTemplate;

    private static final String COLLECTION_NAME = "mission_status";

//    @Value("${zpi.job.mongo.tableName}")
//    private String COLLECTION_NAME;

    private static final String NAME = "count";

    private static final String CHILD_NODES = ".$.childNodes.";

    private static final String BUSINESS_NODES = "businessNodes";

    private static final String DATA_NODES = "dataNodes";

    private static final String BUSINESS_NAME = "businessNodes.name";

    private static final String TOPIC = "TOPIC";

    private static final String PERFORMANCE = "PERFORMANCE";

    private static final String FINISH_STATUS = "FINISH";

    private static final String BEGIN_TIME = "beginTime";

    private static final String END_TIME = "endTime";

    private static final String UPLOAD_NODE = "uploadNode";

    @Override
    public List<MissionStatus> selectAll() {
        return mongoTemplate.findAll(MissionStatus.class);
    }

    @Override
    public MissionStatus create(MissionStatus missionStatus) {
        log.info("func name is [create], input is [{}]", missionStatus);
        return mongoTemplate.save(missionStatus, COLLECTION_NAME);
    }

    @Override
    public void initNodes(MissionStatus status) {
        log.info("func name is [initNodes], input is [{}]", status);
        Query query = new Query(Criteria.where("_id").is(status.get_id()));
        Update update = Update.update("jobUUID", status.getJobUUID())
                .set(DATA_NODES, status.getDataNodes()).set(BUSINESS_NODES, status.getBusinessNodes())
                .set(BEGIN_TIME, status.getBeginTime()).set(END_TIME, status.getEndTIme());
        mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
    }

    /**
     * 分页查询数据状态
     *
     * @param queryVO 分页信息
     * @return 分页信息及结果集
     */
    @Override
    public PageResult<MissionStatusRespVO> getStatusVOPage(MissionStatusQueryVO queryVO) {
        log.info("func name is [getStatusVOPage], input is [{}]", queryVO);
        List<MissionStatusRespVO> statusVOS = selectPageName(queryVO.getPageSize(), queryVO.getPageNum(), queryVO.getName());
        statusVOS.forEach(statusVO -> {
            String uploadNodeStatus = toUploadNodeFunc(Long.parseLong(statusVO.getPkgId())).getStatus();
            if (FINISH_STATUS.equals(uploadNodeStatus)) {
                ProcessInfo dataNodeCount = selectNodeStatusCount(Long.parseLong(statusVO.getPkgId()), NodeType.DATA_NODE.getName());
                ProcessInfo businessNodeCount = selectNodeStatusCount(Long.parseLong(statusVO.getPkgId()), NodeType.BUSINESS_NODE.getName());
                statusVO.setData(dataNodeCount);
                statusVO.setBusiness(businessNodeCount);
            }
            statusVO.setUploadStatus(uploadNodeStatus);
        });
        Page<MissionStatusRespVO> page = new Page<>();

        long total;
        if (StringUtils.hasText(queryVO.getName())) {
            total = selectStatusCount(queryVO.getName()) == null ? 0 : selectStatusCount(queryVO.getName());
        } else {
            total = selectStatusCount(null) == null ? 0 : selectStatusCount(null);
        }
        page.setTotal(total);
        page.setCurrent(queryVO.getPageNum());
        page.setSize(queryVO.getPageSize());
        page.setRecords(statusVOS);
        return new PageResult<>(page);
    }

    /**
     * 根据分页信息返回数据集
     *
     * @param pageSize 每页显示条数
     * @param pageNum  第几页
     * @return 查询到的结果
     */
    private List<MissionStatusRespVO> selectPageName(long pageSize, long pageNum, String name) {

        MatchOperation match = Aggregation.match(Criteria.where("name").regex(".*?" + name + ".*"));

        Aggregation aggregation = StringUtils.hasText(name) ?
                Aggregation.newAggregation(match, Aggregation.project("name", BEGIN_TIME, END_TIME), Aggregation.skip((pageNum - 1) * pageSize), Aggregation.limit(pageSize)) :
                Aggregation.newAggregation(Aggregation.project("name", BEGIN_TIME, END_TIME, UPLOAD_NODE),
                        Aggregation.sort(Sort.Direction.DESC, UPLOAD_NODE + ".uploadTime"),
                        Aggregation.skip(pageNum > 1 ? (pageNum - 1) * pageSize : 0),
                        Aggregation.limit(pageSize));
        List<JSONObject> results = aggregateFuc(aggregation);
        List<MissionStatusRespVO> statuses = new ArrayList<>();
        for (JSONObject mappedResult : results) {
            MissionStatusRespVO statusVO = new MissionStatusRespVO();
            Long id = mappedResult.getLong("_id");
            statusVO.setPkgId(id.toString());
            statusVO.setBeginTime(mappedResult.getLong(BEGIN_TIME));
            statusVO.setEndTime(mappedResult.getLong(END_TIME));
            statusVO.setName(mappedResult.getString("name"));
            statuses.add(statusVO);
        }
        log.info("list mission status is {}", statuses);
        return statuses;
    }

    /**
     * 查询mongo中共有多少条数据
     *
     * @return 总条数
     */
    private Long selectStatusCount(String name) {

        MatchOperation match = Aggregation.match(Criteria.where("name").regex(".*?" + name + ".*"));

        Aggregation aggregation = StringUtils.hasText(name) ?
                Aggregation.newAggregation(match, Aggregation.count().as(NAME)) :
                Aggregation.newAggregation(Aggregation.count().as(NAME));

        List<JSONObject> results = aggregateFuc(aggregation);
        Long count = null;
        if (!CollectionUtils.isEmpty(results)) {
            Integer object = (Integer) results.get(0).get(NAME);
            BigDecimal decimal = new BigDecimal(object);
            count = decimal.longValue();
        }
        return count;
    }

    /**
     * 根据id和节点类型统计节点运行状态
     *
     * @param id   主键id
     * @param type 节点类型
     * @return 统计后的节点运行状态
     */
    private ProcessInfo selectNodeStatusCount(Long id, String type) {
        log.info("func name is [selectNodeStatusCount], input is [{}, {}] ", id, type);
//        [{$unwind:"$dataNodes"},{$group:{"_id":"$dataNodes.status","count":{$sum:1}}}]
        Criteria criteria = Criteria.where("_id").is(id);

        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(criteria),
                Aggregation.unwind(type),
                Aggregation.group(type + ".status").count().as(NAME)
        );
        Map<String, Integer> map = new HashMap<>();

        if (type.equals(NodeType.BUSINESS_NODE.getName())) {
            Criteria criter = Criteria.where("_id").is(id);
            Aggregation aggre = Aggregation.newAggregation(
                    Aggregation.match(criter),
                    Aggregation.unwind(type),
                    Aggregation.group(type + ".childNodes.general.status").count().as(NAME)
            );
            List<JSONObject> jsonObjects = aggregateFuc(aggre);
            for (JSONObject mappedResult : jsonObjects) {
                int count = mappedResult.getInteger(NAME);
                String status = mappedResult.getString("_id");
                if (StringUtils.hasText(status)) {
                    map.put(status, count);
                }
            }
        }

        List<JSONObject> results = aggregateFuc(aggregation);
        log.info("[selectNodeStatusCount] results is {}", results);

        for (JSONObject mappedResult : results) {
            int count = mappedResult.getInteger(NAME);
            String status = mappedResult.getString("_id");
            map.merge(status, count, Integer::sum);
        }
        log.info("status count map is {}", map);

        int init = map.get(JobStatus.INIT.toString()) == null ? 0 : map.get(JobStatus.INIT.toString());
        int failed = map.get(JobStatus.FAILED.toString()) == null ? 0 : map.get(JobStatus.FAILED.toString());
        int succeeded = map.get(FINISH_STATUS) == null ? 0 : map.get(FINISH_STATUS);
        int running = map.get(JobStatus.RUNNING.toString()) == null ? 0 : map.get(JobStatus.RUNNING.toString());
        int total = init + failed + succeeded + running;

        ProcessInfo info = new ProcessInfo();
        info.setInit(init);
        info.setFailed(failed);
        info.setSucceeded(succeeded);
        info.setRunning(running);
        info.setTotal(total);

        log.info("process info is {}", info);

        return info;
    }

    /**
     * 查询上传节点信息
     *
     * @param id 主键id
     * @return 上传节点信息
     */
    private UploadNode toUploadNodeFunc(Long id) {
        List<JSONObject> obj = selectNodeByIdAndType(id, NodeType.UPLOAD_NODE.getName());
        Object object = obj.get(0).get(NodeType.UPLOAD_NODE.getName());
        String str = JSON.toJSONString(object);
        return JsonUtils.jsonToObject(str, UploadNode.class);
    }

    /**
     * 查询上传节点信息
     *
     * @param id 主键id
     * @return 上传节点信息
     */
    @Override
    public UploadNode selectUploadNode(Long id) {
        log.info("func name is [selectUploadNode], input is [id={}]", id);
        return toUploadNodeFunc(id);
    }

    /**
     * 查询节点信息
     *
     * @param statusQueryVO 主键id，节点类型
     * @return 节点集合
     */
    @Override
    public List<NodeInfo> listNodeInfo(StatusQueryVO statusQueryVO) {
        log.info("func name is [listNodeInfo], input is [{}]", statusQueryVO);
        return toNodeInfoFunc(statusQueryVO.getId(), statusQueryVO.getType());
    }

    /**
     * 分页查询节点信息
     *
     * @param statusQueryVO 分页信息，主键id，节点类型
     * @return 分页后的节点信息列表
     */
    @Override
    public PageResult<NodeInfo> pageNodeInfo(StatusQueryVO statusQueryVO) {
        log.info("func name is [pageNodeInfo], input is [{}]", statusQueryVO);

        List<NodeInfo> nodeInfos = selectNodePageByIdAndType(statusQueryVO);
//        List<NodeInfo> nodeInfos = toNodeInfoFunc(statusQueryVO.getId(), statusQueryVO.getType());
        Page<NodeInfo> page = new Page<>();
        page.setTotal(selectDataNodesCount(statusQueryVO.getId()));
        page.setCurrent(statusQueryVO.getPageNum());
        page.setSize(statusQueryVO.getPageSize());
        page.setRecords(nodeInfos);
        return new PageResult<>(page);
    }

    /**
     * 根据id查询数据处理节点的总条数
     *
     * @param id 主键id
     * @return 数据处理节点总条数
     */
    private Long selectDataNodesCount(Long id) {
        Criteria criteria = Criteria.where("_id").is(id);
        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(criteria),
                Aggregation.unwind(DATA_NODES),
                Aggregation.count().as(NAME)
        );
        List<JSONObject> results = aggregateFuc(aggregation);
        if (CollectionUtils.isEmpty(results)) {
            return 0L;
        }
        Integer object = (Integer) results.get(0).get(NAME);
        BigDecimal decimal = new BigDecimal(object);
        return decimal.longValue();
    }

    /**
     * 根据id和类型查询node节点信息集合
     *
     * @param id   主键id
     * @param type 节点类型
     * @return 节点集合
     */
    private List<NodeInfo> toNodeInfoFunc(Long id, String type) {
        if (!type.equals(NodeType.BUSINESS_NODE.getName()) && !type.equals(NodeType.DATA_NODE.getName())) {
            log.error("type is not correct, check type!");
            return new ArrayList<>();
        }
        List<JSONObject> objects = selectNodeByIdAndType(id, type);
        if (CollectionUtils.isEmpty(objects)) {
            return new ArrayList<>();
        }
        return (List<NodeInfo>) objects.get(0).get(type);
    }

    /**
     * 根据id和类型查询node节点信息
     *
     * @param id       主键id
     * @param nodeType 节点类型
     * @return 结果集
     */
    private List<JSONObject> selectNodeByIdAndType(Long id, String nodeType) {
        Criteria criteria = Criteria.where("_id").is(id);
        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(criteria),
                Aggregation.project(nodeType)
        );
        return aggregateFuc(aggregation);
    }

    private List<NodeInfo> selectNodePageByIdAndType(StatusQueryVO queryVO) {
        if (!"dataNodes".equals(queryVO.getType())) {
            return new ArrayList<>();
        }
        Criteria criteria = Criteria.where("_id").is(queryVO.getId());
        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(criteria),
                Aggregation.project(queryVO.getType()),
                Aggregation.unwind(queryVO.getType()),
                Aggregation.skip((queryVO.getPageNum() - 1) * queryVO.getPageSize()),
                Aggregation.limit(queryVO.getPageSize())
        );
        List<JSONObject> objects = aggregateFuc(aggregation);
        if (CollectionUtils.isEmpty(objects)) {
            return new ArrayList<>();
        }
        List<NodeInfo> list = new ArrayList<>();
        for (JSONObject object : objects) {
            NodeInfo nodeInfo = JSON.parseObject(JSON.toJSONString(object.get("dataNodes")), NodeInfo.class);
            list.add(nodeInfo);
        }
//        return (List<NodeInfo>) objects.get(0).get(queryVO.getType());
        return list;
    }


    /**
     * aggregation调用mongoTemplate的公共方法
     *
     * @param aggregation 聚合函数
     * @return 结果集
     */
    private List<JSONObject> aggregateFuc(Aggregation aggregation) {
        AggregationResults<JSONObject> aggregate = mongoTemplate.aggregate(aggregation, COLLECTION_NAME, JSONObject.class);
        List<JSONObject> results = aggregate.getMappedResults();
        log.info("results is {}", results);
        return results;
    }

    @Override
    public NodeInfo selectNodeInfo(Long id, String name) {
        List<NodeInfo> nodeInfos = toNodeInfoFunc(id, DATA_NODES);
        if (!CollectionUtils.isEmpty(nodeInfos)) {
            List<NodeInfo> collect = nodeInfos.stream().filter(nodeInfo -> StringUtils.hasText(nodeInfo.getName())).collect(Collectors.toList());
            return collect.stream().filter(nodeInfo -> name.equals(nodeInfo.getName())).collect(Collectors.toList()).get(0);
        }
        return null;
    }

    @Override
    public void updateNodeByType(MissionStatusResult result) {
        String nodeName;
        switch (result.getType()) {
            case "VIDEO":
                nodeName = DATA_NODES;
                updateNodeStatus("VIDEO", result.getRawName(), result.getPkgId());
                updateRawChildNode(result, nodeName);
                return;
            case "FUSE":
                nodeName = DATA_NODES;
                updateNodeStatus("FUSE", result.getRawName(), result.getPkgId());
                updateRawChildNode(result, nodeName);
                return;
            case "PERCEPTION":
                nodeName = DATA_NODES;
                updateNodeStatus("PERCEPTION", result.getRawName(), result.getPkgId());
                updateRawChildNode(result, nodeName);
                return;
            case PERFORMANCE:
            case TOPIC:
                updateNodeInfo(result);
                return;
            case "GENERAL":
                nodeName = BUSINESS_NODES;
                updateTopicChildNode(result, nodeName);
                return;
            case "UPLOAD":
                uploadUploadNode(result);
                return;
            case "PREPROCESS":
                nodeName = "uploadNode";
                updateUploadChildNode(result, nodeName);
            default:
                return;
        }
    }

    private void updateNodeStatus(String type, String rawName, Long id) {
        NodeInfo nodeInfo = queryNodesByIdAndName(id, rawName);
        if (!ObjectUtils.isEmpty(nodeInfo)) {
            String status = nodeInfo.getChildNodes().get(type).getStatus();
            log.info("children node status is {}", status);
            Query query = new Query(Criteria.where("_id").is(id).and("dataNodes.name").is(rawName));
            Update update = new Update();
            if (!ObjectUtils.isEmpty(nodeInfo) && "INIT".equals(status)) {
                update.set("dataNodes.$.childNodes." + type + ".status", JobStatus.RUNNING.toString());
                mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
            }
        }
    }

    private void updateRawChildNode(MissionStatusResult result, String nodeName) {

        NodeInfo nodeInfo = queryNodesByIdAndName(result.getPkgId(), result.getRawName());
        if (ObjectUtils.isEmpty(nodeInfo)) {
            return;
        }
        if (!result.getJobId().equals(nodeInfo.getJobId())) {
            return;
        }

        Query query = new Query(Criteria.where("_id").is(result.getPkgId()).and(nodeName + ".name").is(result.getRawName()));

        Update update = new Update();
        if (nodeInfo.getStatus().equals("INIT")) {
            update.set(nodeName + ".$.status", JobStatus.RUNNING);
        }

        if (result.getEndTime() != null && result.getEndTime() != 0) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".endTime", result.getEndTime());
        }
        if (StringUtils.hasText(result.getStatus())) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".status", result.getStatus());
        }
        if (!ObjectUtils.isEmpty(result.getPodInfo())) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".podInfo", result.getPodInfo());
        }
        if (result.getProgress() != 0) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".progress", result.getProgress());
        }
        if (!ObjectUtils.isEmpty(result.getPointLog())) {
            update.push(nodeName + CHILD_NODES + result.getType() + ".dotLogs", result.getPointLog());
        }
        if (StringUtils.hasText(result.getJobId())) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".jobId", result.getJobId());
        }
        if (result.getDuration() != null && result.getDuration() != 0) {
            update.set(nodeName + CHILD_NODES + result.getType() + ".duration", result.getDuration());
        }
        if (result.getStartTime() != null && result.getStartTime() != 0) {
            if (!ObjectUtils.isEmpty(nodeInfo) && ObjectUtils.isEmpty(nodeInfo.getBeginTime())) {
                Query q = new Query(Criteria.where("_id").is(result.getPkgId()).and(nodeName + ".name").is(result.getRawName()));
                Update u = new Update();
                u.set(nodeName + ".$.beginTime", result.getStartTime());
                mongoTemplate.updateFirst(q, u, COLLECTION_NAME);
            }
            update.set(nodeName + CHILD_NODES + result.getType() + ".beginTime", result.getStartTime());
        }
        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }

        NodeInfo info = queryNodesByIdAndName(result.getPkgId(), result.getRawName());

        String status = null;
        Long nowTime = 0L;
        List<String> statusList = new ArrayList<>();
        if (!ObjectUtils.isEmpty(info) && !CollectionUtils.isEmpty(info.getChildNodes())) {
            for (NodeInfo value : info.getChildNodes().values()) {
                statusList.add(value.getStatus());
            }
        }

        List<String> collect = statusList.stream().filter(str -> str.equals("INIT") || str.equals("RUNNING")).collect(Collectors.toList());
        if (collect.size() == 0) {
            List<String> failed = statusList.stream().filter(str -> str.equals("FAILED")).collect(Collectors.toList());
            if (failed.size() != 0) {
                status = "FAILED";
            } else {
                status = FINISH_STATUS;
            }
            nowTime = new Date().getTime() / 1000;
        }

        if (StringUtils.hasText(status)) {
            Query que = new Query(Criteria.where("_id").is(result.getPkgId()).and(nodeName + ".name").is(result.getRawName()));
            Update upd = new Update();
            upd.set(nodeName + ".$.status", status).set(nodeName + ".$.endTime", nowTime);
            mongoTemplate.updateFirst(que, upd, COLLECTION_NAME);
        }
    }


    private void updateTopicChildNode(MissionStatusResult result, String nodeName) {
        if (!checkStatusByJob(result.getPkgId(), result.getJobId(), nodeName)) {
            return;
        }
        Query query = new Query(Criteria.where("_id").is(result.getPkgId()).and(nodeName + ".name").is("TOPIC"));

        Update update = buildUpdate(result, nodeName);

        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
    }

    private void updateUploadChildNode(MissionStatusResult result, String nodeName) {
        Query query = new Query(Criteria.where("_id").is(result.getPkgId()));

        Update update = buildUpdate(result, nodeName);

        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
    }



    private Update buildUpdate(MissionStatusResult result, String nodeName) {
        String child;
        if (nodeName.equals("uploadNode")) {
            child = ".childNodes.";
        }else {
            child = ".$.childNodes.";
        }
        Update update = new Update();
        if (result.getProgress() != 0) {
            // uploadNode.$.childNodes.preprocess.progress
            update.set(nodeName + child + result.getType().toLowerCase() + ".progress", result.getProgress());
        }
        if (!ObjectUtils.isEmpty(result.getPodInfo())) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".podInfo", result.getPodInfo());
        }
        if (result.getDuration() != null && result.getDuration() != 0) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".duration", result.getDuration());
        }
        if (!ObjectUtils.isEmpty(result.getPointLog())) {
            update.push(nodeName + child + result.getType().toLowerCase() + ".dotLogs", result.getPointLog());
        }
        if (StringUtils.hasText(result.getStatus())) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".status", result.getStatus());
        }
        if (result.getStartTime() != null && result.getStartTime() != 0) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".beginTime", result.getStartTime());
        }
        if (StringUtils.hasText(result.getJobId())) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".jobId", result.getJobId());
        }
        if (result.getEndTime() != null && result.getEndTime() != 0) {
            update.set(nodeName + child + result.getType().toLowerCase() + ".endTime", result.getEndTime());
        }
        return update;
    }


    private NodeInfo queryNodesByIdAndName(Long id, String name) {

        List<NodeInfo> nodeInfos = toNodeInfoFunc(id, DATA_NODES);
        NodeInfo nodeInfo = null;
        if (!CollectionUtils.isEmpty(nodeInfos)) {
            for (NodeInfo info : nodeInfos) {
                if (name.equals(info.getName())) {
                    log.info("[queryNodesByIdAndName] node info is {}", info);
                    nodeInfo = info;
                }
            }
        }
        return nodeInfo;
    }

    private boolean checkStatusByJob(Long id, String jobId, String nodeName) {
        boolean boo = false;
        Criteria criteria = Criteria.where("_id").is(id).and(nodeName + ".jobId").is(jobId);
        Aggregation aggregation = Aggregation.newAggregation(
                Aggregation.match(criteria)
        );
        List<JSONObject> jsonObjects = aggregateFuc(aggregation);
        if (!CollectionUtils.isEmpty(jsonObjects)) {
            boo = true;
        }
        return boo;
    }

    private void updateNodeInfo(MissionStatusResult result) {
        if (!checkStatusByJob(result.getPkgId(), result.getJobId(), BUSINESS_NODES)) {
            return;
        }
        Query query = new Query(Criteria.where("_id").is(result.getPkgId()).and(BUSINESS_NAME).is(result.getType()));
        Update update = new Update();

        if (result.getStartTime() != null && result.getStartTime() != 0) {
            update.set(BUSINESS_NODES + ".$.beginTime", result.getStartTime());
        }
        if (result.getEndTime() != null && result.getEndTime() != 0) {
            update.set(BUSINESS_NODES + ".$.endTime", result.getEndTime());
        }
        if (result.getDuration() != null && result.getDuration() != 0) {
            update.set(BUSINESS_NODES + ".$.duration", result.getDuration());
        }
        if (!ObjectUtils.isEmpty(result.getPodInfo())) {
            update.set(BUSINESS_NODES + ".$.podInfo", result.getPodInfo());
        }
        if (StringUtils.hasText(result.getJobId())) {
            update.set(BUSINESS_NODES + ".$.jobId", result.getJobId());
        }
        if (!ObjectUtils.isEmpty(result.getPointLog())) {
            update.push(BUSINESS_NODES + ".$.dotLogs", result.getPointLog());
        }
        if (result.getProgress() != 0) {
            update.set(BUSINESS_NODES + ".$.progress", result.getProgress());
        }
        if (StringUtils.hasText(result.getStatus())) {
            update.set(BUSINESS_NODES + ".$.status", result.getStatus());
        }

        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
    }

    private void uploadUploadNode(MissionStatusResult result) {
        Query query = new Query(Criteria.where("_id").is(result.getPkgId()));
        Update update = new Update();
        if (result.getDuration() != null && result.getDuration() != 0) {
            update.set("uploadNode.duration", result.getDuration());
        }
        if (!ObjectUtils.isEmpty(result.getPodInfo())) {
            update.set("uploadNode.podInfo", result.getPodInfo());
        }
        if (result.getProgress() != 0) {
            update.set("uploadNode.progress", result.getProgress());
        }
        if (StringUtils.hasText(result.getStatus())) {
            update.set("uploadNode.status", result.getStatus());
        }
        if (result.getStartTime() != null && result.getStartTime() != 0) {
            update.set("uploadNode.beginTime", result.getStartTime());
            update.set("uploadNode.uploadTime", result.getStartTime());
        }
        if (result.getEndTime() != null && result.getEndTime() != 0) {
            update.set("uploadNode.endTime", result.getEndTime());
            update.set("uploadNode.uploadTime", result.getEndTime());
        }
        if (StringUtils.hasText(result.getJobId())) {
            update.set("uploadNode.jobId", result.getJobId());
        }
        if (!ObjectUtils.isEmpty(result.getPointLog())) {
            update.push("uploadNode.dotLogs", result.getPointLog());
        }

        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
    }


    @Override
    public void initSingleBusinessWorkflow(String templateName, Long id, NodeInfo nodeInfo) {
        Query query = new Query();
        Update update = new Update();
        if (PERFORMANCE.equals(templateName)) {
            query = new Query(Criteria.where("_id").is(id).and(BUSINESS_NAME).is(PERFORMANCE));
            update.set(BUSINESS_NODES + ".$", nodeInfo);
        }
        if (TOPIC.equals(templateName)) {
            query = new Query(Criteria.where("_id").is(id).and(BUSINESS_NAME).is(TOPIC));
            update.set(BUSINESS_NODES + ".$", nodeInfo);
        }
        if (!ObjectUtils.isEmpty(update)) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }

        log.info("init businessNodes, name is [{}]", templateName);
    }

    @Override
    public void initSingleMFPWorkflow(String rawName, Long id, NodeInfo nodeInfo) {
        Query query = new Query(Criteria.where("_id").is(id).and(DATA_NODES + ".name").is(rawName));
        nodeInfo.setName(rawName);
        Update update = new Update();
        update.set(DATA_NODES + ".$", nodeInfo);
        if (!ObjectUtils.isEmpty(update)) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
        log.info("init dataNodes, name is [{}]", rawName);
    }

    @Override
    public void updateBigBagEndTime(Long id, Long endTime) {
        Query query = new Query(Criteria.where("_id").is(id));
        Update update = Update.update(END_TIME, endTime / 1000);
        mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
    }

    @Override
    public void insertUploadTime(Long id) {
        UploadNode uploadNode = toUploadNodeFunc(id);
        Long time = null;
        if (!ObjectUtils.isEmpty(uploadNode)) {
            time = ObjectUtils.isEmpty(uploadNode.getEndTime()) ? uploadNode.getBeginTime() : uploadNode.getEndTime();
        }
        Query query = new Query(Criteria.where("_id").is(id));
        Update update = new Update();
        if (time!=null) {
            update.set("uploadNode.uploadTime", time);
        }
        if (!update.equals(new Update())) {
            mongoTemplate.updateFirst(query, update, COLLECTION_NAME);
        }
    }

    @Override
    public void removeById(Long id) {
        Query query = new Query(Criteria.where("_id").is(id));
        DeleteResult remove = mongoTemplate.remove(query, COLLECTION_NAME);
        log.info("remove result is {}", remove);
    }

    @Override
    public void updateDataNodeById(Long id) {
        // select dataNodes list
        List<NodeInfo> nodeInfos = toNodeInfoFunc(id, DATA_NODES);

        List<NodeInfo> newDataNodes = new ArrayList<>();
        for (NodeInfo info : nodeInfos) {
            NodeInfo childNode = info.getChildNodes().get(NodeStatus.VIDEO.getName());

            Map<String, NodeInfo> map = new HashMap<>();
            map.put(NodeStatus.VIDEO.getName(), childNode);

            info.setEndTime(childNode.getEndTime());
            info.setStatus("FINISH");
            info.setChildNodes(map);
            newDataNodes.add(info);
        }

        Query query = new Query(Criteria.where("_id").is(id));
        Update update = new Update();
        update.set("dataNodes", newDataNodes);
        mongoTemplate.updateFirst(query, update, COLLECTION_NAME);

    }

}
